package com.mjlf.mqserver.server;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.mjlf.mqserver.entity.Message;
import com.mjlf.mqserver.entity.Topic;
import com.mjlf.mqserver.entity.TopicBluck;
import com.mjlf.mqserver.inter.MessageFactory;
import com.mjlf.mqserver.inter.ServerFactory;
import com.mjlf.mqzook.inter.TopicFactroy;
import com.mjlf.restclient.entity.RestParams;
import com.mjlf.restclient.template.RestTemplate;
import com.sun.org.apache.regexp.internal.RE;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

/**
 * @ClassName MessageServer
 * @Author mjlft
 * @Date 2020/2/19 12:39
 * @Version 1.0
 * @Description TODO
 */
@Service
public class MessageServer implements MessageFactory {

    @Autowired
    private RestTemplate restTemplate;
    @Value("${client.host}")
    private String host;
    @Value("${server.port}")
    private Integer port;

    @Override
    public void insertMessage(String topicName, String message) throws Exception {
        JSONObject jsonObject = TopicBluck.TOPIC.getJSONObject("data").getJSONObject(topicName);
        if(jsonObject != null){
            if(jsonObject.getString("leader").equals(host+":"+port)){
                Message msg = new Message();
                msg.setMessage(message);
                msg.setInsertTime(System.currentTimeMillis());

                List<Message> messages = MessageFactory.message.get(topicName);
                if(messages == null){
                    messages = new ArrayList<>();
                }
                messages.add(msg);
                MessageFactory.message.put(topicName, messages);
            }else {
                RestParams params = new RestParams();
                params.setUrl("http://"+jsonObject.getString("leader")+"/insertMessage");
                JSONObject p = new JSONObject();
                p.put("topicName", topicName);
                p.put("message", message);
                params.setParams(p.toJSONString());
                restTemplate.post(params);
            }
        }
    }

    @Override
    public String getMessage(String topicName, String groupName) throws Exception {
        JSONObject jsonObject = TopicBluck.TOPIC.getJSONObject("data").getJSONObject(topicName);
        if(jsonObject != null){
            if(jsonObject.getString("leader").equals(host+":"+port)){
                Integer index = MessageFactory.customerIndex.get(groupName);
                if(index == null){
                    index = 0;
                }
                List<Message> messages = MessageFactory.message.get(topicName);
                if(messages == null || messages.isEmpty()){
                    return "";
                }
                if(index >= messages.size()){
                    return "读完了";
                }
                Message message = messages.get(index);
                index += 1;
                MessageFactory.customerIndex.put(groupName, index);
                return message.getMessage();
            }else {
                RestParams params = new RestParams();
                params.setUrl("http://"+jsonObject.getString("leader")+"/getMessage");
                JSONObject p = new JSONObject();
                p.put("topicName", topicName);
                p.put("groupName", groupName);
                params.setParams(p.toJSONString());
                return restTemplate.get(params);
            }
        }
        return "异常";
    }

    @Override
    public void synMessage() throws Exception {
        JSONObject topics = TopicBluck.TOPIC.getJSONObject("data");
        if(topics == null){
            return;
        }
        Iterator<String> iterator = topics.keySet().iterator();
        while(iterator.hasNext()){
            String topicName = iterator.next();
            JSONObject topic = topics.getJSONObject(topicName);
            //判断重节点
            if((host+":"+port).equals(topic.getString("follow"))){
                List<Message> messages = MessageFactory.copyMessage.get(topicName);
                int index = 0;
                if(messages == null){
                    messages = new ArrayList<>();
                    MessageFactory.copyMessage.put(topicName, messages);
                }else {
                    index = messages.size();
                }
                String url = "http://" + topic.getString("leader") + "/synMessage";
                RestParams restParams = new RestParams();
                restParams.setUrl(url);
                JSONObject parm = new JSONObject();
                parm.put("index", index);
                parm.put("topicName", topicName);
                restParams.setParams(parm.toJSONString());
                JSONObject jsonObject = JSONObject.parseObject(restTemplate.post(restParams));

                if(jsonObject.containsKey("data")){
                    JSONArray jsonArray = jsonObject.getJSONArray("data");
                    for(int i = 0; i < jsonArray.size(); i ++){
                        JSONObject msg = jsonArray.getJSONObject(i);
                        Message message = new Message();
                        message.setInsertTime(msg.getLong("insertTime"));
                        message.setMessage(msg.getString("message"));
                        messages.add(message);
                    }
                }
            }
        }
    }

    @Override
    public void synCustomerIndex() throws Exception {
        for(String server : ServerFactory.servers){
            String url = "http://" + server + "/synConsumerIndex";
            RestParams restParams = new RestParams();
            restParams.setUrl(url);

            JSONObject jsonObject = JSONObject.parseObject(restTemplate.get(restParams));
            if(jsonObject.containsKey("data")){
                JSONObject cusomerIndex = jsonObject.getJSONObject("data");
                Iterator<String> iterator = cusomerIndex.keySet().iterator();
                while (iterator.hasNext()){
                    String groupName = iterator.next();
                    Integer index = cusomerIndex.getInteger(groupName);

                    if(MessageFactory.customerIndex.containsKey(groupName)){
                        if(index > MessageFactory.customerIndex.get(groupName)){
                            MessageFactory.customerIndex.put(groupName, index);
                        }
                    }else {
                        MessageFactory.customerIndex.put(groupName, index);
                    }
                }
            }
        }
    }
}
